#pragma once

#include "detail.hpp"
#include "fields.hpp"
#include "abstract.hpp"
#include "message.hpp"
#include <mutex>
#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/Buffer.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/net/TcpClient.h>

namespace rpc
{
    class MuduoBuffer : public Basebuffer
    {
    public:
        using ptr = std::shared_ptr<MuduoBuffer>;
        MuduoBuffer(muduo::net::Buffer *buf):_buf(buf){}
        virtual size_t readableSize() override//返回当前可读取数据的字节数
        {
            return _buf->readableBytes();
        }
        virtual int32_t peekInt32() override//读取4字节的数据
        {
            return _buf->peekInt32();
        }
        virtual void retriveInt32() override//跳过4字节的数据
        {
            return _buf->retrieveInt32();
        }
        virtual int32_t readInt32() override//读取并跳过4字节的数据
        {
            return _buf->readInt32();//里面包含网络字节序转换为主机字节序的过程
        }
        virtual std::string retriveAsString(size_t len) override//读取len字节的数据
        {
            return _buf->retrieveAsString(len);
        }
    private:
        muduo::net::Buffer *_buf;
    };
    class BufferFactory
    {
    public:
        template<class ...Args>
        static Basebuffer::ptr create(Args&&... args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    class LVProtocol : public BaseProtocol
    {
    public:
        using ptr = std::shared_ptr<BaseProtocol>;
        //|----len----|----------len bytes----------|
        //|----len----|Mtype---IDlength---MID---body|
        virtual std::string serialize(const BaseMessage::ptr& msg) 
        {
            std::string body = msg->serialize();
            std::string id = msg->id();
            int32_t mtype = htonl((int32_t)msg->mtype());
            int32_t idlen = htonl(id.size());//从主机字节序转换为网络字节序

            int32_t h_total_len = mtypeFieldsLength + idlenFieldsLength + id.size() + body.size();
            int32_t n_total_len = htonl(h_total_len);

            std::string result;
            result.reserve(h_total_len);//预先开辟空间，防止频繁的扩容
            result.append((char*)&n_total_len,lenFieldsLength);
            result.append((char*)&mtype,mtypeFieldsLength);
            result.append((char*)&idlen,idlenFieldsLength);
            result.append(id);
            result.append(body);
            return result;
        }
        virtual bool canProcessed(const Basebuffer::ptr& buffer)//判断缓冲区中的数据是否足够一条消息的处理
        {
            if(buffer->readableSize() < lenFieldsLength)
            return false;
            int total_len = buffer->peekInt32() + lenFieldsLength;
            if(buffer->readableSize() < total_len)
            {
                return false;
            }
            return true;
        }
        virtual bool onMessage(const Basebuffer::ptr& buffer,BaseMessage::ptr& msg)
        {
            if(canProcessed(buffer))
            {
                int32_t total_len = buffer->readInt32();//读取消息总长度
                Mtype mtype = (Mtype)buffer->readInt32();//读取消息数据类型
                int32_t id_len = buffer->readInt32();//读取id的长度
                int32_t body_len = total_len - mtypeFieldsLength - idlenFieldsLength - id_len;
                std::string id = buffer->retriveAsString(id_len);
                std::string body = buffer->retriveAsString(body_len);
                //DLOG("%s",body.c_str());
                msg = MessageFactory::create(mtype);
                if(msg.get() == nullptr)
                {
                    ELOG("消息类型错误，构造消息对象失败！");
                    return false;
                }
                bool ret = msg->unserialize(body);
                if(ret == false)
                {
                    ELOG("消息正文反序列化失败！");
                    return false;
                }
                msg->setid(id);
                msg->setmtype(mtype);
                return true;
            }
            return false;
        }
    private:
        const size_t lenFieldsLength = 4;
        const size_t mtypeFieldsLength = 4;
        const size_t idlenFieldsLength = 4;
    };

    class ProtocolFactory
    {
    public:
        template<class ...Args>
        static BaseProtocol::ptr create(Args&& ...args)
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
        }
    };



    class MuduoConnection : public BaseConnection
    {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;
        MuduoConnection(const muduo::net::TcpConnectionPtr &conn,const BaseProtocol::ptr &protocol) 
        : _protocol(protocol),
        _conn(conn)
        {}
        virtual void send(const BaseMessage::ptr &msg) override
        {
            std::string message = _protocol->serialize(msg);
            _conn->send(message);
        }
        virtual void shutdown() override
        {
            _conn->shutdown();
        }
        virtual bool connected() override
        {
            return _conn->connected();
        }
    private:
        BaseProtocol::ptr _protocol;
        muduo::net::TcpConnectionPtr _conn;
    };

    class ConnectionFactory
    {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;
        template<class ...Args>
        static MuduoConnection::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };
    
    class MuduoServer : public BaseServer
    {
    public:
        using ptr = std::shared_ptr<MuduoServer>;
        MuduoServer(int port) :
         _svr(&_baseloop,muduo::net::InetAddress("0.0.0.0",port),"MuduoServer",muduo::net::TcpServer::kReusePort),
         _protocol(ProtocolFactory::create())
        {}
        virtual void start() override
        {
            _svr.setConnectionCallback(std::bind(&MuduoServer::onConnection,this,std::placeholders::_1));
            _svr.setMessageCallback(std::bind(&MuduoServer::onMessage, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _svr.start();
            _baseloop.loop();
        }
    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if(conn->connected())
            {
                DLOG("连接成功");
                auto muduo_conn = ConnectionFactory::create(conn,_protocol);
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    _conns.insert(std::make_pair(conn, muduo_conn));
                }
                if (_cb_connection) _cb_connection(muduo_conn);
            }
            else 
            {
                DLOG("断开连接");
                BaseConnection::ptr muduo_conn;
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    auto it = _conns.find(conn);
                    if(it == _conns.end())
                    {
                        DLOG("连接不存在");
                        return;
                    }
                    DLOG("删除该连接");
                    muduo_conn = it->second;
                    _conns.erase(conn);
                }
                if (_cb_close) _cb_close(muduo_conn);
                return;
            }
        }
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
        {
            DLOG("连接有数据到来");
            std::shared_ptr<Basebuffer> base_buf = BufferFactory::create(buf);
            while(1)
            {
                if(_protocol->canProcessed(base_buf) == false)
                {
                    if(base_buf->readableSize() > maxDataSize)
                    {
                        conn->shutdown();
                        ELOG("缓冲区数据过大");
                        return ;
                    }
                    DLOG("数据量不足！");
                    break;
                }
                DLOG("缓冲区中数据可处理！");
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(base_buf,msg);
                if(ret == false)
                {
                    conn->shutdown();
                    ELOG("缓冲区中数据错误");
                    return;
                }
                DLOG("消息反序列化成功！");
                BaseConnection::ptr base_conn;
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    std::unordered_map<muduo::net::TcpConnectionPtr,BaseConnection::ptr>::iterator it= _conns.find(conn);
                    if(it == _conns.end())
                    {
                        conn->shutdown();
                        return;
                    }
                    base_conn = it->second;
                }
                DLOG("调用回调函数进行消息处理！");
                if (_cb_msg) _cb_msg(base_conn, msg);
            }
        }

    private:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _svr;
        std::mutex _mtx;
        std::unordered_map<muduo::net::TcpConnectionPtr,BaseConnection::ptr> _conns;
    };

    class ServerFactory
    {
    public:
        template<class ...Args>
        static std::shared_ptr<MuduoServer> create(Args&& ...args)
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };

    class MuduoClient : public BaseClient
    {
    public:
        using ptr = std::shared_ptr<MuduoClient>;
        MuduoClient(const std::string& sip,uint16_t sport) :
         _protocol(ProtocolFactory::create()),
         _baseloop(_loopthread.startLoop()),
         _downlatch(1),
         _client(_baseloop,muduo::net::InetAddress(sip,sport),"MuduoClient")//这里的ip地址和port是服务器地址
        {}
        virtual void connect() override
        {
            DLOG("设置回调函数，连接服务器");
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection,this,std::placeholders::_1));
            _client.setMessageCallback(std::bind(&MuduoClient::onMessage,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _client.connect();
            _downlatch.wait();
            DLOG("连接服务器成功！");
        }
        virtual void shutdown() override
        {
            _client.disconnect();
        }
        virtual bool send(const BaseMessage::ptr &msg) override
        {
            if(connected() == false)
            {
                ELOG("连接已断开！");
                return false;
            }
            _conn->send(msg);
            return true;
        }
        virtual BaseConnection::ptr connection() override
        {
            return _conn;
        }
        virtual bool connected()
        {
            return (_conn && _conn->connected());
        }
    private:
    void onConnection(const muduo::net::TcpConnectionPtr &conn) 
    {
        if(conn->connected())
        {
            DLOG("建立连接成功");
            _downlatch.countDown();
            _conn = ConnectionFactory::create(conn,_protocol);
        }
        else 
        {
            DLOG("连接断开");
            _conn.reset();
        }
    }
    void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
    {
        DLOG("连接有数据到来，开始处理！");
        std::shared_ptr<Basebuffer> base_buf = BufferFactory::create(buf);

        while(1)
        {
            if(_protocol->canProcessed(base_buf) == false)
            {
                if(base_buf->readableSize() > maxDataSize)
                {
                    conn->shutdown();
                    DLOG("缓冲区数据过大");
                    return;
                }
                break;
            }
            else 
            {
                DLOG("缓冲区中数据足够处理");
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(base_buf,msg);
                if(ret == false)
                {
                    conn->shutdown();
                    DLOG("缓冲区中数据错误");
                    return;
                }
                if(_cb_msg) _cb_msg(_conn,msg);
            }
        }
    }
    private:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        BaseConnection::ptr _conn;
        muduo::CountDownLatch _downlatch;
        muduo::net::EventLoopThread _loopthread;
        muduo::net::EventLoop *_baseloop;
        muduo::net::TcpClient _client;
    };

    class CilentFactory
    {
    public:
        template<class ...Args>
        static BaseClient::ptr create(Args ...args)
        {
            return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
        }
    };
}
